package com.lyrcsoft.rxjava1x.operators.convert;

import rx.Observable;
import rx.schedulers.Schedulers;

/**
 * @author fujc2dev@126.com
 * @date 2024-11-20
 */
public class Rx1groupBy {
    public static void main(String[] args) throws InterruptedException {
        Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9)
                .groupBy(number -> number % 2 == 0 ? "Even" : "Odd")
                .subscribe(groupedObservable -> {
                    groupedObservable.reduce(0, (sum, item) -> sum + item) // 对每组累加
                            .subscribe(total ->
                                    System.out.println("组 [" + groupedObservable.getKey() + "] 的总和: " + total));
                });

        System.out.println("-----------------------------------------------------------------------------------------");
        Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9)
                .groupBy(number -> number % 2 == 0 ? "Even" : "Odd")
                .subscribe(groupedObservable -> {
                    groupedObservable
                            .observeOn(Schedulers.computation()) // 每组独立线程处理
                            .subscribe(number ->
                                    System.out.println("线程: " + Thread.currentThread().getName()
                                            + " | 组 [" + groupedObservable.getKey() + "] 的值: " + number));
                });

        Thread.sleep(1000); // 等待异步任务完成

    }
}
